package finalProject;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.SequenceFileAsTextInputFormat;

public class Test {
	

	static ArrayList<Path> sendList = new ArrayList<Path>();
	private static HashSet<NounPair> annotatedTrue;
	private static HashSet<NounPair> annotatedFalse;
	
	
	private static void readAnnotated() throws IOException {
		annotatedTrue = new HashSet<NounPair>();
		annotatedFalse = new HashSet<NounPair>();
		URL filePate = new URL("http://www.cs.bgu.ac.il/~dsp142/wiki.files/hypernym.txt");
		BufferedReader in = new BufferedReader(new InputStreamReader(filePate.openStream()));

		String inputLine;
		while ((inputLine = in.readLine()) != null)
		{
			String[] input = inputLine.split("\t");
			if(input.length>=3){
				if(input[2].equals("True") == true)
				{
					annotatedTrue.add(new NounPair(input[0], input[1]));
				}
				else
				{
					annotatedFalse.add(new NounPair(input[0], input[1]));
				}
			}
		}
		in.close();
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		
try{
	NounPair p1 = new NounPair("comoros", "territory");
	
	readAnnotated();
	if(annotatedTrue.contains(p1))
		System.out.println("y");
		String uri = "biarcs.30-of-99";
	    File file = new File(uri);
	    BufferedReader br = new BufferedReader(new FileReader(file));
	    String line;
	    while ((line = br.readLine()) != null && sendList.size()<100000) {
	       Map( new Text(line));
	    }

	
	//	WritableComparable key = (WritableComparable) reader.getKeyClass().newInstance();
	//	Writable value = (Writable) reader.getValueClass().newInstance();
		//while (reader.next(key, value))
		 // Map((LongWritable)key, (Text)value);
		
		Collections.sort(sendList);
		int count = 0;
		ArrayList<Path> toSend = new ArrayList<Path>();
		Path last = null;
		for (int i = 0 ; i < sendList.size() ; i++) {
			Path p = sendList.get(i);
		//	if(last != null && last.compareTo(p) == 0)
			//	continue;
			if(last == null || last.getPath().compareTo(p.getPath()) == 0){
				toSend.add(p);
			}
			else
			{
				Reduce(toSend);
				toSend.clear();
				toSend.add(p);
			}
			last = p;
		}
			
			
			
		
			br.close();			
		} catch (Exception e) {
			// TODO: handle exception
		}
		

	}

	
	private static void Reduce(ArrayList<Path> values) {
		int dpMin = 1;
		HashSet<NounPair> cached = new HashSet<NounPair>();
		NounPair last = null;
		for (Path p: values) {
			NounPair nounPair = p.getNounPair();
			if(cached.size() < dpMin)
			{
				cached.add(nounPair);
			}
			else
			{
				if(last.equals(nounPair) == false)
				{
					//context.write(nounPair, key.getPath());
					System.out.println("Pair=" + nounPair+ " Path=" + p.getPath());
				}
			}
			last = nounPair;
		}
		if(cached.size() == dpMin)
		{
			for (NounPair nounPair : cached) {
				//context.write(nounPair, key.getPath());
				System.out.println("Pair=" + nounPair+ " Path=" + values.get(0).getPath());
			}
		}

		
	}


	public static void Map(Text value)
	{
		String[] parts = value.toString().split("\t");
		if (parts.length < 2) {
			return;
		}
		ArrayList<Token> tokens = createTokens(parts[1].split(" "));
		if (countNouns(tokens) < 2) {
			return;
		}

		Tree t = new Tree(tokens);

		for(int sIndex = 0; sIndex < tokens.size() ; sIndex++) {
			Token source = tokens.get(sIndex);
			if(source.isNoun() == true)
			{
				for(int tIndex = sIndex + 1 ; tIndex < tokens.size() ; tIndex++)
				{
					Token target = tokens.get(tIndex);
					if(target.isNoun() == true)
					{
						String path = t.getPath(source, target);
						if (path == null)
						{
							continue;
						}
						NounPair nounObj = new NounPair(source.getWord(), target.getWord());
						Path pathObj = new Path(new Text(path), nounObj);
						sendList.add(pathObj);
					}
				}
			}
		}
	}
	
	private static ArrayList<Token> createTokens(String[] parts) {
		ArrayList<Token> ans = new ArrayList<Token>();
		for (String string : parts) {
			ans.add(new Token(string));
		}
		return ans;
	}

	private static int countNouns(ArrayList<Token> tokens) {
		int nounCounter = 0;
		for (Token token : tokens) {
			if (token.isNoun())
				nounCounter++;
		}
		return nounCounter;
	}
}
